///////////////////////////////////////////////
// C++ Delayed Queue template (Support cancel task)
// Author: Geng Xin (sportgary@foxmail.com)
// https://gitee.com/sportgary/delayed-queue-cpp

#ifndef CD788B8A_68EB_40FC_BCF1_93BCBF67F471
#define CD788B8A_68EB_40FC_BCF1_93BCBF67F471

#include <chrono>
#include <condition_variable>
#include <mutex>
#include <utility>
#include <vector>
#include <algorithm>
#include <functional>

// using namespace std::chrono_literals;


namespace XUtils {

using Clock = std::chrono::steady_clock;
using TimePoint = typename Clock::time_point;


template <typename VTYPE>
class DQTask {
public:
    uint64_t   taskid;
    VTYPE      val;
    TimePoint  time;
} ;


// _taskid 从 1 开始递增。所以 0 可以用来表示无效值
template <typename T>
class DelayedQueue {

public:

    DelayedQueue() = default;
    explicit DelayedQueue(std::size_t initial_capacity)
    {
        _queue.reserve(initial_capacity);
    }
    ~DelayedQueue() = default;

    DelayedQueue(const DelayedQueue&) = delete;
    DelayedQueue& operator=(const DelayedQueue&) = delete;

    uint64_t put(T v, TimePoint tp = std::chrono::steady_clock::now())
    {
        if (_stopped) {
            return 0;
        }

        DQTask<T> task;
        task.taskid = _taskid;
        task.time = tp;
        task.val = v;

        std::lock_guard<decltype(_mtx)> lk(_mtx);

        _queue.push_back(std::move(task));
        // 按时间排序
        std::sort(begin(_queue), end(_queue), [](DQTask<T>&& a, DQTask<T>&& b) { return a.time > b.time; });

        _cv.notify_one();

        return _taskid++;
    }

    // 不考虑时间的版本，直接插入到队列尾部
    uint64_t put(T v)
    {
        if (_stopped) {
            return 0;
        }

        DQTask<T> task;
        task.taskid = _taskid;
        task.time = std::chrono::steady_clock::now();
        task.val = v;

        std::lock_guard<decltype(_mtx)> lk(_mtx);

        _queue.push_back(std::move(task));

        _cv.notify_one();

        return _taskid++;
    }


    bool take(T &val)
    {
        std::unique_lock<decltype(_mtx)> lk(_mtx);
        auto now = Clock::now();
        // wait condition: (empty && stop) || (!empty && back.tp <= now)
        // while (!(_queue.empty() && _stopped) && !(!_queue.empty() && _queue.back().time <= now)) {
        while (!( _stopped) && !(!_queue.empty() && _queue.back().time <= now)) {
            if (_queue.empty())
                _cv.wait(lk);
            else
                _cv.wait_until(lk, _queue.back().time);
            now = Clock::now();
        }

        if (_stopped) {
            return false; // invalid value
        }

        val = std::move(_queue.back().val);
        _queue.pop_back();
        if (_queue.empty() && _stopped)
            _cv.notify_all();

        return true;
    }

    void stop()
    {
        std::lock_guard<decltype(_mtx)> lk(_mtx);
        _stopped = true;
        _queue = {};
        _cv.notify_all();
    }

    size_t size()
    {
        return _queue.size();
    }

    bool empty()
    {
        return _queue.empty();
    }

    void cancel_task(uint64_t taskid)
    {
        std::lock_guard<decltype(_mtx)> lk(_mtx);

        // std::cout << "++ q.size=" << _queue.size() << std::endl;

        _queue.erase( std::remove_if(std::begin(_queue), std::end(_queue), [taskid](const DQTask<T>& task) {
                return task.taskid == taskid;
            }),
            std::end(_queue) );

        // std::cout << "-- q.size=" << _queue.size() << std::endl;
    }

    // 不停止循环，只清空队列
    void clear() {
        std::unique_lock<decltype(_mtx)> lk(_mtx);
        _queue.clear();
        _cv.notify_all();
    }


    void clear_if( std::function<bool(const DQTask<T>&)> pred ) {
        std::unique_lock<decltype(_mtx)> lk(_mtx);
        _queue.erase( std::remove_if(std::begin(_queue), std::end(_queue), [&pred](const DQTask<T>& task) {
                return pred(task);
            }),
            std::end(_queue) );
    }

private:
    std::vector< DQTask<T> >  _queue;
    std::condition_variable  _cv;
    uint64_t   _taskid = 1;
    volatile bool       _stopped = false;
    std::mutex _mtx;
};

} // namespace XUtils


#endif /* CD788B8A_68EB_40FC_BCF1_93BCBF67F471 */
